package com.example.dayu521.sg_oss.controller.msg;

import com.alibaba.fastjson2.JSONObject;
import com.example.dayu521.sg_oss.domain.Trade;
import com.example.dayu521.sg_oss.domain.donain_mid.CartParam;
import com.example.dayu521.sg_oss.domain.donain_mid.TradeParam;
import com.example.dayu521.sg_oss.service.Kl;
import com.example.dayu521.sg_oss.service.PSettleService;
import com.example.dayu521.sg_oss.service.TradeService;
import com.example.dayu521.sg_oss.service.feign.CartService;
import io.seata.core.context.RootContext;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageExt;
import org.example.dayu521.ORExecption;
import org.example.dayu521.OrRes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.concurrent.ArrayBlockingQueue;

import static java.lang.StringTemplate.STR;

@Component("settleTransactionListener")
@Slf4j
public class SettleTransactionListener implements TransactionListener {

    @Autowired
    private TradeService tradeService;

    @Autowired
    private PSettleService pSettleService;

    @Autowired
    private StreamBridge streamBridge;

    @Value("${sg-oss.order.timeout_level}")
    private Long timeout_level;
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {

        var fr = (Kl.Px<TradeParam, ArrayBlockingQueue<OrRes<?>>, Exception>) arg;
        var tradeParam = fr.getP();

        try {
            //增加字段,订单号
            pSettleService.order_trade(tradeParam);
        } catch (Exception e) {
            fr.setE(e);
            fr.getR().add(OrRes.error());
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }

        var one = tradeService.lambdaQuery().eq(Trade::getSession_id, tradeParam.getSettle_id()).one();

        //fixme 此处如果失败,会由事务会查进行重试
        //延迟消息 rocketmq5 https://rocketmq.apache.org/zh/docs/featureBehavior/02delaymessage/
        //rocketmq4 https://rocketmq.apache.org/zh/docs/4.x/producer/04message3/
        //这里是4版本
        // TODO: 11/23/23 需要增加一个截止日期字段
        var builder = MessageBuilder.withPayload(one.getBusiness_id());
        builder.setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL,timeout_level);
        streamBridge.send("order-out-0",builder.build());

        //fixme 这里失败,只能由用户自己去订单中心查看
        var xds=STR."dda";
        fr.getR().add(OrRes.ok(one.getBusiness_id()));
        return LocalTransactionState.COMMIT_MESSAGE;

        //org.springframework.messaging.Message的header,调用msg的putProperty方法
        //同时,特定于rocketmq的属性,都放到属性中
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        var s = JSONObject.parseObject(new String(msg.getBody()), String.class);
        var one = tradeService.lambdaQuery().eq(Trade::getSession_id,  s).one();
        if(one==null){
            log.warn("事务仍未检测到");
            return LocalTransactionState.UNKNOW;
        }if(one.getState()==0){
            log.info("事务暂未确认");
            return LocalTransactionState.UNKNOW;
        } else if (one.getState() == 1) {
            var builder = MessageBuilder.withPayload(one.getBusiness_id());
            builder.setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL,4);//30s
            streamBridge.send("order-out-0",builder.build());
            return LocalTransactionState.COMMIT_MESSAGE;
        }else if (one.getState() == 2) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }else{
            log.error("交易状态不正确,请检查是否人为修改了数据 {}",msg.getBody().toString());
            return LocalTransactionState.UNKNOW;
        }

    }

}